package com.py.custom.business.demo.core.mq;

import com.alibaba.fastjson.JSON;
import com.py.custom.business.demo.core.mapstruct.BeanConvertUtils;
import com.py.custom.business.demo.core.mq.task.TaskRunDTO;
import com.py.custom.business.demo.core.utils.MessageProcessUtils;
import com.py.custom.business.demo.protocol.dto.MessageTaskDTO;
import com.py.custom.business.demo.protocol.request.MessageRequest;
import com.py.custom.common.web.utils.ApplicationContextUtils;
import com.py.custom.data.entity.demo.Record;
import com.py.custom.data.entity.demo.Template;
import com.py.custom.data.mapper.demo.RecordDao;
import com.py.custom.data.mapper.demo.TemplateDao;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * description：kafka消费者；todo：待接入
 */
@Component
public class CustomKafkaConsumer {

    @Resource
    private TemplateDao templateDao;

    @Resource
    private RecordDao recordDao;

    @Autowired
    private BeanConvertUtils beanConvertUtils;

    /**
     * todo 改造成自定义的线程池
     */
    private ExecutorService executorService = Executors.newSingleThreadExecutor();

    public void consumer(String strRecord) {
        MessageRequest request = JSON.parseObject(strRecord, MessageRequest.class);
        Template template = templateDao.selectById(request.getTemplateId());
        if (Objects.isNull(template)) {
            return;
        }

        // 1.处理请求，获取数据 任务消息类；
        String content = MessageProcessUtils.process(template.getVariableValue(), template.getTemplateContent(),
                request.getVariableMap());
        MessageTaskDTO messageTaskDto = beanConvertUtils.message2Dto(content, request);

        // 2.获取任务
        TaskRunDTO task = ApplicationContextUtils.getBean(TaskRunDTO.class);
        if (Objects.isNull(task)) {
            return;
        }
        // 3.插入 任务记录
        Record record = beanConvertUtils.template2Record(request, 0, content, null);
        record.setMessageId("uuid");
        recordDao.insert(record);

        // 4.设置消息任务
        messageTaskDto.setTaskId(record.getRecordId());
        task.setMessageTaskDto(messageTaskDto);

        // 5.执行
        executorService.execute(task);
    }
}
